package com.atguigu.app.dwd;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.DWDStaticConstants;
import com.atguigu.common.ODSStaticConstants;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 * 数据  ： web/app -> nginx -> springboot -> kafka -> flink ->kafka
 * 程序    mock  ->nginx ->logger.sh ->kafka -> BaseLogApp ->kafka
 *
 */

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /**
         * 生产环境下一定要写的，不过学习阶段还要开启hdfs比较麻烦先关闭
         *         System.setProperty("HADOOP_USER_NAME","atguigu");
         *         env.setStateBackend(new FsStateBackend(OdsStaticConstants.CHECK_POINT_URL));
         *         // 设置checkpoint时间
         *         env.enableCheckpointing(5000L);
         *         // 设置精准一次性
         *         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
         *         // 设置check超时时间
         *         env.getCheckpointConfig().setCheckpointTimeout(10000L);
         *         // 设置最多两个checkpoint一块
         *         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
         *         // 设置两个check间隔
         *         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
         */
        // 消费kafka数据
        DataStreamSource<String> dataStreamSource =
                env.addSource(KafkaUtil.getKafkaConsumer(ODSStaticConstants.CDC_TO_ODS_KAFKA_TOPIC_LOG, DWDStaticConstants.ODS_DWD_GROUP_ID));
        // 将异常信息输出到测输出流
        OutputTag<String> dirtyOutput = new OutputTag<String>("Dirty") {};
        SingleOutputStreamOperator<JSONObject> jsonObjectSingleOutputStreamOperator = dataStreamSource
                .process(new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, Context context, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSONObject.parseObject(value);
                            collector.collect(jsonObject);
                        } catch (Exception e) {
                            context.output(dirtyOutput, value);
                        }
                    }
                });
        SingleOutputStreamOperator<JSONObject> objectSingleOutputStreamOperator = jsonObjectSingleOutputStreamOperator
                .keyBy(f -> f.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {
                    private ValueState<JSONObject> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("mid", JSONObject.class));
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
                        if ("1".equals(value.getJSONObject("common").getString("is_new"))) {
                            if (valueState.value() != null) {
                                value.getJSONObject("common").put("is_new", "0");
                            } else {
                                valueState.update(value);
                            }
                        }
                        return value;
                    }
                });
        OutputTag<String> startOut = new OutputTag<String>("start") {};
        OutputTag<String> display = new OutputTag<String>("display") {};
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = objectSingleOutputStreamOperator.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {
                String start = jsonObject.getString("start");
                if (start != null && start.length() > 0) {
                    context.output(startOut, start);
                } else {
                    collector.collect(jsonObject.toJSONString());

                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject j = displays.getJSONObject(i);
                            j.put("page_id", pageId);
                            context.output(display, j.toJSONString());
                        }
                    }
                }
            }
        });
        DataStream<String> startOutput = stringSingleOutputStreamOperator.getSideOutput(startOut);
        DataStream<String> displayOutput = stringSingleOutputStreamOperator.getSideOutput(display);
        DataStream<String> dirtyOutSidePut = jsonObjectSingleOutputStreamOperator.getSideOutput(dirtyOutput);
        // 分流结果输出出去
        startOutput.addSink(KafkaUtil.getKafkaProducter(DWDStaticConstants.DWD_KAFKA_TOPIC_START_LOG));
        displayOutput.addSink(KafkaUtil.getKafkaProducter(DWDStaticConstants.DWD_KAFKA_TOPIC_DISPLAY_LOG));
        stringSingleOutputStreamOperator.addSink(KafkaUtil.getKafkaProducter(DWDStaticConstants.DWD_KAFKA_TOPIC_PAGE_LOG));

        env.execute();
    }
}
